package com.pine.common.threadpool.core;

import com.pine.common.threadpool.DaemonThreadFactory;
import com.pine.common.threadpool.RejectedPolicyFactory;
import com.pine.common.threadpool.ThreadPoolConfig;
import com.pine.common.threadpool.ThreadPoolConstant;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 动态线程池刷新(仅支持刷新：最大线程数、核心线程数、非核心线程存活时间、拒绝策略)
 *
 * @author pine manage
 * @since 2024-08-28
 */
@Getter
@Slf4j
public class DynamicThreadPoolRefresh {

    /**
     * 刷新线程池参数定时任务
     */
    private final static ScheduledExecutorService REFRESH_THREAD_POOL_PARAM_SCHEDULED_EXECUTOR_SERVICE =
            new ScheduledThreadPoolExecutor(1, new DaemonThreadFactory("refresh-thread-pool-param-scheduled-thread-pool"));

    /**
     * 刷新线程池参数定时任务初始延迟时间
     */
    private final long scheduleInitialDelay;
    /**
     * 刷新线程池参数定时任务执行间隔时间
     */
    private final long schedulePeriod;
    /**
     * 刷新线程池参数定时任务执行间隔时间单位
     */
    private final TimeUnit scheduleUnit;

    public DynamicThreadPoolRefresh() {
        this(60L, 60L, TimeUnit.SECONDS);
    }

    public DynamicThreadPoolRefresh(long scheduleInitialDelay, long schedulePeriod, TimeUnit scheduleUnit) {
        this.scheduleInitialDelay = scheduleInitialDelay;
        this.schedulePeriod = schedulePeriod;
        this.scheduleUnit = scheduleUnit;
    }

    /**
     * 刷新
     */
    public void refresh() {
        // 刷新线程池参数
        this.refreshThreadPoolParam();
    }

    /**
     * 刷新线程池参数
     */
    private void refreshThreadPoolParam() {
        // 延迟60s，然后每隔60s执行一次刷新
        REFRESH_THREAD_POOL_PARAM_SCHEDULED_EXECUTOR_SERVICE.scheduleAtFixedRate(() -> {
            // 获取当前所有的线程池map
            Map<String, ThreadPoolExecutor> threadPoolExecutorMap = DynamicThreadPoolManager.getThreadPoolExecutorMap();
            if (threadPoolExecutorMap == null || threadPoolExecutorMap.isEmpty()) {
                return;
            }

            threadPoolExecutorMap.forEach((configKey, threadPoolExecutor) -> {
                try {
                    String[] businessCodeAndKey = configKey.split(ThreadPoolConstant.THREAD_POOL_NAME_SEPARATOR);
                    ThreadPoolConfig threadPoolConfig = ThreadPoolConfigFactory.getThreadPoolConfig(businessCodeAndKey[0], businessCodeAndKey[1]);

                    if (threadPoolConfig == null) {
                        return;
                    }
                    log.info("线程池{}配置信息={}", configKey, threadPoolConfig);
                    log.info("刷新线程池{}参数，原参数：MaximumPoolSize={},CorePoolSize={},KeepAliveTime={}s,RejectedPolicy={}", configKey, threadPoolExecutor.getMaximumPoolSize(), threadPoolExecutor.getCorePoolSize(), threadPoolExecutor.getKeepAliveTime(TimeUnit.SECONDS), threadPoolExecutor.getRejectedExecutionHandler().getClass().getSimpleName());

                    // 最大线程数更新
                    setIfTrue(!Objects.equals(threadPoolExecutor.getMaximumPoolSize(), threadPoolConfig.getMaximumPoolSize()),
                            () -> threadPoolExecutor.setMaximumPoolSize(threadPoolConfig.getMaximumPoolSize()));
                    // 核心线程数更新
                    setIfTrue(!Objects.equals(threadPoolExecutor.getCorePoolSize(), threadPoolConfig.getCorePoolSize()),
                            () -> threadPoolExecutor.setCorePoolSize(threadPoolConfig.getCorePoolSize()));
                    // 非核心线程存活时间更新
                    setIfTrue(!Objects.equals(threadPoolExecutor.getKeepAliveTime(threadPoolConfig.getKeepAliveTimeUnit()), threadPoolConfig.getKeepAliveTime()),
                            () -> threadPoolExecutor.setKeepAliveTime(threadPoolConfig.getKeepAliveTime(), threadPoolConfig.getKeepAliveTimeUnit()));
                    // 拒绝策略更新
                    setIfTrue(!Objects.equals(threadPoolExecutor.getRejectedExecutionHandler(), RejectedPolicyFactory.getRejectedPolicy(threadPoolConfig.getRejectedPolicyCode())),
                            () -> threadPoolExecutor.setRejectedExecutionHandler(RejectedPolicyFactory.getRejectedPolicy(threadPoolConfig.getRejectedPolicyCode())));
                } catch (Exception e) {
                    log.warn("刷新线程池{}参数失败", configKey, e);
                }
            });
        }, scheduleInitialDelay, schedulePeriod, scheduleUnit);
    }


    static void setIfTrue(boolean condition, Runnable runnable) {
        if (condition) {
            runnable.run();
        }
    }
}
